package com.xf.day07_graphx

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
/**
 * 操作属性图
 *     图的顶点个数， 图的边个数
 *     图的入度， 图的出度，  图的入度和出度之和
 *     图的缓存
 *    节点属性 和 边属性 的变换;  三元视图
 *
 */
object OperatorGraph {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("testGraphx").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")

    // 创建一个顶点集的 RDD,  VertexId 一个 Long 类型, 顶点属性是一个 二元组
    val users : RDD[( VertexId, (String, Int))] = sc.parallelize(
      Array(
        ( 1L, ("菲菲", 22)),
        ( 2L, ("乔乔", 17)),
        ( 3L, ("王强", 19)),
        ( 4L, ("王刚", 21)),
        ( 5L, ("李倩", 20)),
        ( 6L, ("林峰", 25))
    ))

    // 创建一个边集的 RDD
    val relationships :RDD[ Edge[Int]] = sc.parallelize(
      Array( Edge(1L ,3L, 15) ,
             Edge(2L ,1L, 10) ,
             Edge(2L ,3L, 21) ,
             Edge(2L ,4L, 22) ,
             Edge(3L ,6L, 10) ,
             Edge(4L ,5L, 17) ,
             Edge(5L ,6L, 20)
      ))

    // 定义边中 用户缺失时的 默认 (缺失) 用户
    val defaultUser = ("某某", 18)

    // 使用 users 和 relationships 两个RDD 实例化 Graph类建立一个 Graph对象
    val userGraph = Graph( users, relationships, defaultUser)

    /**
     *  获取 属性图中的 边的数量
     *  图中边的数量为: 7
     */
    val edgeCnt : Long = userGraph.numEdges
    println("图中边的数量为: " +  edgeCnt )
    println("=======================>")

    /**
     * 获取 属性图中的 顶点的数量
     * 图中顶点数量为: 6
     */
    val verticesCnt = userGraph.numVertices
    println( "图中顶点数量为: " +  verticesCnt)
    println("=======================>")

    /**
     * 获取属性图 中的 所有顶点的 入度
     * ( 顶点的名字, 顶点的入度 )
     *
     * (1,1)
     * (3,2)
     * (4,1)
     * (5,1)
     * (6,2)
     */
    userGraph.inDegrees.collect().foreach( println)
    println("=======================>")

    /**
     *  获取属性图 中的 所有顶点的 出度
     *
     * (1,1)
     * (2,3)
     * (3,1)
     * (4,1)
     * (5,1)
     */
    userGraph.outDegrees.collect().foreach(println)
    println("=======================>")

    /**
     * 获取属性图 中的 所有顶点的 入度 和 出度 之和
     *
     * (1,2)
     * (2,3)
     * (3,3)
     * (4,2)
     * (5,2)
     * (6,2)
     */
    userGraph.degrees.collect().foreach(println)
    println("=======================>")

    /**
     * 直接查看顶点信息
     *
     *    (1,(菲菲,22))
     *    (2,(乔乔,17))
     *    (3,(王强,19))
     *    (4,(王刚,21))
     *    (5,(李倩,20))
     *    (6,(林峰,25))
     */
    userGraph.vertices.collect().foreach( println)
    println("=======================>")

    /**
     * 通过元组下标的方式  ===> 查看顶点信息
     *
     * (菲菲,22,1)
     * (乔乔,17,2)
     * (王强,19,3)
     * (王刚,21,4)
     * (李倩,20,5)
     * (林峰,25,6)
     */
    // RDD[( VertexId, (String, Int))]
    userGraph.vertices.map{  v => ( v._2._1 , v._2._2, v._1) }.collect().foreach(println)
    println("=======================>")

    /**
     * 增加过滤条件的 顶点视图
     *
     * (2,(乔乔,17))
     * (3,(王强,19))
     */
    userGraph.vertices.filter{ case (id, (name, age )) => age < 20 }.collect().foreach(println)
    println("=======================>")

    /**
     * 查看所有边的信息
     *
     *    Edge(1,3,15)
     *    Edge(2,1,10)
     *    Edge(2,3,21)
     *    Edge(2,4,22)
     *    Edge(3,6,10)
     *    Edge(4,5,17)
     *    Edge(5,6,20)
     */
    userGraph.edges.collect().foreach( println )
    println("=======================>")

    /**
     * 通过模式匹配 并增加过滤条件
     *    Edge(2,1,10)
     */
    userGraph.edges.filter{ case Edge( src , dst, attr) => src > dst }.collect().foreach( println )
    println("=======================>")

    /**
     * 通过下标 查看 边字段
     *
     * 源点 ID: 1 ,目的点 ID: 3 ,边属性: 15
     * 源点 ID: 2 ,目的点 ID: 1 ,边属性: 10
     * 源点 ID: 2 ,目的点 ID: 3 ,边属性: 21
     * 源点 ID: 2 ,目的点 ID: 4 ,边属性: 22
     * 源点 ID: 3 ,目的点 ID: 6 ,边属性: 10
     * 源点 ID: 4 ,目的点 ID: 5 ,边属性: 17
     * 源点 ID: 5 ,目的点 ID: 6 ,边属性: 20
     */
    userGraph.edges.map{
      v => ("源点 ID: " + v.srcId + " ,目的点 ID: " + v.dstId + " ,边属性: " + v.attr )
    }.collect().foreach( println)
    println("=======================>")

    /**
     * 查看 边点三元组 视图  => 输出 所有边的 三元组
     *
     * ((1,(菲菲,22)),(3,(王强,19)),15)
     * ((2,(乔乔,17)),(1,(菲菲,22)),10)
     * ((2,(乔乔,17)),(3,(王强,19)),21)
     * ((2,(乔乔,17)),(4,(王刚,21)),22)
     * ((3,(王强,19)),(6,(林峰,25)),10)
     * ((4,(王刚,21)),(5,(李倩,20)),17)
     * ((5,(李倩,20)),(6,(林峰,25)),20)
     */
    userGraph.triplets.collect().foreach( println)
    println("=======================>")

    /**
     *  查看 边点三元组 视图  ==> 通过下标查询
     *    (菲菲,王强,15)
     *    (乔乔,菲菲,10)
     *    (乔乔,王强,21)
     *    (乔乔,王刚,22)
     *    (王强,林峰,10)
     *    (王刚,李倩,17)
     *    (李倩,林峰,20)
     */
    userGraph.triplets.map{
      v => ( v.srcAttr._1 , v.dstAttr._1 , v.attr)
    }.collect().foreach( println)
    println("=======================>")

    /**
     * 图的缓存的操作
     */
    userGraph.persist( )
    // userGraph.persist( StorageLevel.DISK_ONLY )
    println("=======================>")
    /**
     *  默认保存在 内存中
     */
    userGraph.cache()
    println("=======================>")

    /**
     *  释放图缓存
     */
    userGraph.unpersist( blocking = true )
    println("=======================>")

    /**
     *  对 mapVertices 顶点属性 进行变换  ==> 得到一个 新图
     *
     * (1,菲菲)
     * (2,乔乔)
     * (3,王强)
     * (4,王刚)
     * (5,李倩)
     * (6,林峰)
     */
    val new_userGraph : Graph[String, Int] = userGraph.mapVertices( (VertexID, ID) => ID._1 )
    new_userGraph.vertices.collect().foreach( println )  // 输出所有顶点
    println("=======================>")

    /**
     *  对 mapEdges 对 边属性进行 变换
     *    Edge(1,3,150)
     *    Edge(2,1,100)
     *    Edge(2,3,210)
     *    Edge(2,4,220)
     *    Edge(3,6,100)
     *    Edge(4,5,170)
     *    Edge(5,6,200)
     */
    val new_edges_userGraph = userGraph.mapEdges(e => e.attr * 10)
    new_edges_userGraph.edges.collect().foreach( println )
    println("=======================>")


    /**
     * 对 边属性 mapTriplets 进行 变换
     *    Edge(1,3,菲菲15王强)
     *    Edge(2,1,乔乔10菲菲)
     *    Edge(2,3,乔乔21王强)
     *    Edge(2,4,乔乔22王刚)
     *    Edge(3,6,王强10林峰)
     *    Edge(4,5,王刚17李倩)
     *    Edge(5,6,李倩20林峰)
     */
    val new_userGraph2 = userGraph.mapTriplets(triplet => triplet.srcAttr._1  +  triplet.attr +  triplet.dstAttr._1)
    new_userGraph2.edges.collect().foreach( println )  // 输出所有边

  }

}
